package com.shujia.dw

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.table.catalog.hive.HiveCatalog
import org.apache.flink.types.Row

object OdsETLToDwdUser {
  def main(args: Array[String]): Unit = {
    val bsEnv = StreamExecutionEnvironment.getExecutionEnvironment
    val bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()

    val bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings)

    val hiveCatalog = new HiveCatalog("myHive", "sent", "SentimentCompute/src/main/resources")

    bsTableEnv.registerCatalog("myHive", hiveCatalog)
    bsTableEnv.useCatalog("myHive")

    val userTable = bsTableEnv.sqlQuery(
      """
        |select distinct
        |  md5(cast(id as String)) as id
        |  ,case when gender = 'f' then '女'
        |   when gender = 'm' then '男'
        |   end as gender
        |  ,screen_name
        |  ,followers_count
        |  ,follow_count
        |from ods.ods_kafka_user
        |""".stripMargin)

    val userDS = userTable.toRetractStream[Row]
    val table = userDS.toTable(bsTableEnv, $"flag", $"row").select($"row")
    bsTableEnv.createTemporaryView("tmp", table)
    bsTableEnv.executeSql(
      """
        |insert into dwd.dwd_kafka_user_msk
        |select
        |  id
        |  ,gender
        |  ,screen_name
        |  ,followers_count
        |  ,follow_count
        |from tmp
        |""".stripMargin)


    bsEnv.execute()


  }

}
